﻿#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <queue>
#include <memory>
#include <atomic>
#include <mutex>
#include <functional>
#include <condition_variable>
#include <thread>
#include <unordered_map>
#include <future>

class Any
{
public:
    Any()  = default;
    ~Any() = default;

    Any(Any&&)            = default;
    Any& operator=(Any&&) = default;

    Any(const Any&)             = delete;
    Any& operator=(const Any&)  = delete;

    // 使Any可以接受任意类型数据
    template<typename T>
    Any(T data) : upBase_(std::make_unique<Derive<T>>(data)) {}

    // 将data数据提取出来
    template<typename T>
    T cast_()
    {
        Derive<T>* pd = dynamic_cast<Derive<T>*>(upBase_.get());
        if (!pd)
        {
            throw "type is unmatch";
        }

        return pd->data_;
    }

private: 
    // 基类
    class Base
    {
    public:
        virtual ~Base() = default;
    };

    // 子类
    template<typename T>
    class Derive : public Base
    {
    public:
        Derive(T data) : data_(data) {}
        T data_;    // 保存了任意的类型
    };

private:
    std::unique_ptr<Base> upBase_;
};

// 信号量
class Semaphore
{
public:
    Semaphore(int limit = 0) : resLimit_(limit) {}
    ~Semaphore() = default;

    // 获取一个信号量资源
    void wait()
    {
        std::unique_lock<std::mutex> lock(mtx_);
        cond_.wait(lock, [this]() { return resLimit_ > 0; });
        --resLimit_;
    }

    // 增加一个信号量资源
    void post()
    {
        std::unique_lock<std::mutex> lock(mtx_);
        ++resLimit_;
        cond_.notify_all();
    }

private:
    int resLimit_;
    std::mutex mtx_;
    std::condition_variable cond_;
};

class Task;
// 提交到线程池的任务执行完成后的返回值类型
class Result
{
public:
    Result(std::shared_ptr<Task> spTask, bool isValid = true);
    ~Result() = default;

    void setVal(Any any);

    // 获取task的返回值
    Any get();

private:
    Any any_;   // 存储任务的返回值
    Semaphore sem_; 
    std::shared_ptr<Task> task_; // 指向对应获取返回值的任务对象
    std::atomic_bool isValid_;   // 返回值是否有效
};

// 任务抽象基类
class Task
{
public:
    Task();
    ~Task() = default;

    void exec();
    void setResult(Result* pRes);

    // 用户可以自定义任意任务类型，从Task继承，重写run方法，实现自定义任务处理
    virtual Any run() = 0;

private:
    Result* pResult_;
};

// 线程池支持的模式
enum class PoolMode
{
    MODE_FIXED,     // 固定数量
    MODE_CACHED,    // 数量可动态增长 适合小而短的任务
};

/**********************************************
// example:
// class MyTask : public Task
// {
// public:
//     virtual Any run() override { //... }
// }
//
// ThreadPool pool;
// pool.start(4);
// pool.submitTask(std::make_shared<MyTask>());
//
***********************************************/
class ThreadPool
{
public:
    ThreadPool();
    ~ThreadPool();

    // 设置线程池模式
    void setMode(PoolMode mode);

    // 开启线程池
    void start(int initThreadCount = std::thread::hardware_concurrency());

    // 设置任务队列上限阈值
    void setTaskQueMaxThreashold(int threashold);

    // 设置线程数量上限阈值
    void setThreadCountMaxThreashold(int threashold);

    // 给线程池提交任务
    Result submitTask(std::shared_ptr<Task> spTask);

    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;

private:
    // 定义线程函数
    void threadFunc();

    // 检查pool的运行状态
    bool checkRunningState() const;

private:
    std::unordered_map<std::thread::id, std::thread> threads_;  // 线程列表
    int initThreadSize_;            // 初始线程数量
    int threadCountMaxThreshold_;   // 线程数量上限
    std::atomic_int curThreadCount_;    // 记录线程数量
    std::atomic_int idleThreadCount_;   // 记录空闲线程的数量

    std::queue<std::shared_ptr<Task>> taskQue_; // 任务队列
    std::atomic_int taskSize_;  // 任务数量
    int taskQueMaxThreshold_;   // 任务队列数量上限

    std::mutex taskQueMtx_; // 保证任务队列的线程安全
    std::condition_variable notFull_;   // 任务队列不满
    std::condition_variable notEmpty_;  // 任务队列不空

    PoolMode poolMode_;     // 当前线程池的工作模式
    std::atomic_bool isRunning_;        // 控制线程是否继续运行
    std::mutex threadMapMutex_; // 用于保护 threads_ 的互斥锁
};

#endif // !THREADPOOL_H
